package producer_test

import (
	"container/list"
	"gitee.com/zaper/kafka-service/model"
	"gitee.com/zaper/private-logger"
	"github.com/petermattis/goid"
	"math/rand"
	"sync"
)

const (
	Producer = iota
	WriterMess
)

var (
	sw          sync.RWMutex
	topicListSw sync.RWMutex
	cusTopicSw  sync.RWMutex
)

func WriterGroup(producerModel *model.KafkaProducerModel) {
	lst := parseTopicsToList(producerModel)
	for i := 0; i < producerModel.ProducerCount; i++ {
		go func() {
			// 随机获取一个生产者类型，开始给自己的主题生产消息
			// 初始化每个协程的信号量
			producerModel.StopChan[goid.Get()] = make(chan struct{})
			factoryProducer()(lst, producerModel)
		}()
	}
}

func factoryProducer() func(*list.List, *model.KafkaProducerModel) {
	// 当前仅有两类生产者(IBM/Segment)，随机获取一类
	num := rand.Intn(2)
	var f func(*list.List, *model.KafkaProducerModel)
	switch num {
	case Producer:
		f = execSegmentProducer
	case WriterMess:
		f = execIBMProducer
	default:
		f = nil
	}
	logger.Infof("Producer function is %d", num)
	return f
}

func parseTopicsToList(producerModel *model.KafkaProducerModel) *list.List {
	topicListSw.Lock()
	defer topicListSw.Unlock()
	lst := list.New()
	for _, topic := range producerModel.Topics {
		lst.PushBack(topic)
	}
	return lst
}

func cusTopicName(lst *list.List, producerModel *model.KafkaProducerModel) string {
	cusTopicSw.Lock()
	defer cusTopicSw.Unlock()
	var topic string
	// 如果主题已经被所有生产者都已经消费了，其他生产者将主题列表里面任意过去主题生产消息
	if lst.Len() == 0 {
		num := rand.Intn(len(producerModel.Topics))
		topic = producerModel.Topics[num]
	} else {
		e := lst.Back()
		topic, _ = e.Value.(string)
		lst.Remove(e)
	}
	return topic
}

func deleteSignMap(m map[int64]chan struct{}, key int64) {
	sw.Lock()
	delete(m, key)
	sw.Unlock()
}
